package hamster.csustef.acquisition.flink.source;


import hamster.csustef.acquisition.api.ElectricApi;
import hamster.csustef.acquisition.constant.AcquisitionConstant;
import hamster.csustef.acquisition.util.TimeUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Date;

/**
 * 生成寝室键数据源
 * 通过工具类来读取寝室键列表文件 = ElectricApi.getRoomIdList()
 * 输出：寝室键 = 127:A113
 *
 * @author yin hamaster
 * @date 2022-06-14
 */
@Slf4j
public class DormKeySource implements SourceFunction<String> {
    boolean running = true;
    int hour = 1000 * 60 * 60;
    int firstTerm = AcquisitionConstant.FIRST_TERM;
    int term = AcquisitionConstant.ELECTRIC_FEE_TERM;
    int gap = AcquisitionConstant.ELECTRIC_FEE_GAP;
    int termNumber;

    public DormKeySource(int runningHour) {
        this.termNumber = runningHour * hour / term;
    }


    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        long startTime = TimeUtil.sleep(firstTerm);

        while (running && startTime <= new Date().getTime() && termNumber > 0) {
            for (String roomId : ElectricApi.getDormKeysList()) {
                sourceContext.collect(roomId);
                Thread.sleep(gap);
            }
            termNumber--;
            startTime = TimeUtil.sleep(term);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

